package appTool

import (
	"fmt"
	"github.com/akutz/memconn"
	"github.com/smallnest/rpcx/client"
	"github.com/smallnest/rpcx/server"
	"net"
	"os"
	"sync"
)

// rpcx直接进程调用不走TCP或者UDP协议,使用内存通信

// 进程内通信地址,结构体
type rpcxInprocessAddr struct{}

func (r rpcxInprocessAddr) Network() string {
	return "memu"
}

func (r rpcxInprocessAddr) String() string {
	return fmt.Sprintf("conn_%d", os.Getpid())
}

type rpcxInprocessDiscovery struct {
	chans []chan []*client.KVPair
	mu    sync.Mutex
	pairs []*client.KVPair
}

func (r *rpcxInprocessDiscovery) setAddress(addr rpcxInprocessAddr) {
	r.pairs = append(r.pairs, &client.KVPair{
		Key:   fmt.Sprintf("%s@%s", addr.Network(), addr.String()),
		Value: "serverMode=all",
	})
}

func (r *rpcxInprocessDiscovery) GetServices() []*client.KVPair {
	return r.pairs
}

func (r *rpcxInprocessDiscovery) WatchService() chan []*client.KVPair {
	r.mu.Lock()
	defer r.mu.Unlock()

	ch := make(chan []*client.KVPair, 10)
	r.chans = append(r.chans, ch)
	return ch
}

func (r *rpcxInprocessDiscovery) RemoveWatcher(ch chan []*client.KVPair) {
	r.mu.Lock()
	defer r.mu.Unlock()

	var chans []chan []*client.KVPair
	for _, c := range r.chans {
		if c == ch {
			continue
		}

		chans = append(chans, c)
	}

	r.chans = chans
}

func (r *rpcxInprocessDiscovery) Clone(servicePath string) (client.ServiceDiscovery, error) {
	return r, nil
}

func (r *rpcxInprocessDiscovery) SetFilter(filter client.ServiceDiscoveryFilter) {
}

func (r *rpcxInprocessDiscovery) Close() {
}

type RpcxInprocess struct {
	addr            rpcxInprocessAddr
	serverListener  net.Listener
	serverDiscovery *rpcxInprocessDiscovery
}

func (r *RpcxInprocess) ServerListener() (net.Listener, error) {
	if r.serverListener == nil {
		listener, err := memconn.Listen(r.addr.Network(), r.addr.String())
		if err != nil {
			return nil, err
		}
		r.serverListener = listener
	}
	return r.serverListener, nil
}

func (r *RpcxInprocess) ServerDiscovery() client.ServiceDiscovery {
	if r.serverDiscovery == nil {
		r.serverDiscovery = &rpcxInprocessDiscovery{
			chans: nil,
			mu:    sync.Mutex{},
			pairs: nil,
		}
		r.serverDiscovery.setAddress(r.addr)
	}
	return r.serverDiscovery
}

func (r *RpcxInprocess) ServerRegisterPlugin() server.Plugin {
	return struct {
	}{}
}
